// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#pragma once

#include <melon/executor.h>
#include <melon/executors/cpu_thread_pool_executor.h>

#include <pollux/common/base/runtime_metrics.h>
#include <pollux/vector/flat_vector.h>
#include <pollux/vector/vector_maker.h>
#include <optional>
#include <pollux/type/cpp_to_type.h>
#include <pollux/vector/vector_fuzzer.h>

namespace kumo::pollux {
    /// Returns indices buffer with sequential values going from size - 1 to 0.
    BufferPtr make_indices_in_reverse(vector_size_t size, memory::MemoryPool *pool);

    BufferPtr make_indices(
        vector_size_t size,
        std::function<vector_size_t(vector_size_t)> indexAt,
        memory::MemoryPool *pool);

    bool is_equal_vectors(const VectorPtr &expected, const VectorPtr &actual, std::string *msg = nullptr);

    // Verify that the values in both 'expected' and 'actual' vectors is the same
    // but only for the rows marked valid in 'rowsToCompare'.
    bool is_equal_vectors(
        const VectorPtr &expected,
        const VectorPtr &actual,
        const SelectivityVector &rowsToCompare, std::string *msg = nullptr);

    /// Verify that 'vector' is copyable, by copying all rows.
    bool is_copyable_vector(const VectorPtr &vector);

    class VectorBuilder {
    protected:
        VectorBuilder() = default;

        ~VectorBuilder();

        static std::shared_ptr<const RowType> make_row_type(
            std::vector<std::shared_ptr<const Type> > &&types) {
            return VectorMaker::row_type(
                std::forward<std::vector<std::shared_ptr<const Type> > &&>(types));
        }

        void set_nulls(
            const VectorPtr &vector,
            std::function<bool(vector_size_t /*row*/)> is_null_at) {
            for (vector_size_t i = 0; i < vector->size(); i++) {
                if (is_null_at(i)) {
                    vector->set_null(i, true);
                }
            }
        }

        static std::function<bool(vector_size_t /*row*/)> null_every(
            int n,
            int startingFrom = 0) {
            return VectorMaker::null_every(n, startingFrom);
        }

        // Helper function for comparing vector results
        template<typename T1, typename T2>
        bool
        compare_values(const T1 &a, const std::optional<T2> &b, std::string &error) {
            bool result = (a == b.value());
            if (!result) {
                error = " " + std::to_string(a) + " vs " + std::to_string(b.value());
            } else {
                error = "";
            }
            return result;
        }

        bool compare_values(
            const StringView &a,
            const std::optional<std::string> &b,
            std::string &error) {
            bool result = (a.getString() == b.value());
            if (!result) {
                error = " " + a.getString() + " vs " + b.value();
            } else {
                error = "";
            }
            return result;
        }

        RowVectorPtr make_row_vector(
            const std::vector<std::string> &childNames,
            const std::vector<VectorPtr> &children,
            std::function<bool(vector_size_t /*row*/)> is_null_at = nullptr) {
            auto row_vector = vectorMaker_.row_vector(childNames, children);
            if (is_null_at) {
                set_nulls(row_vector, is_null_at);
            }
            return row_vector;
        }

        RowVectorPtr make_row_vector(
            const std::vector<VectorPtr> &children,
            std::function<bool(vector_size_t /*row*/)> is_null_at = nullptr) {
            auto row_vector = vectorMaker_.row_vector(children);
            if (is_null_at) {
                set_nulls(row_vector, is_null_at);
            }
            return row_vector;
        }

        RowVectorPtr make_row_vector(
            const std::shared_ptr<const RowType> &row_type,
            vector_size_t size) {
            return vectorMaker_.row_vector(row_type, size);
        }

       RowVectorPtr make_row_vector(
            const RowTypePtr &type,
            const VectorFuzzer::Options &fuzzerOpts) {
            VectorFuzzer fuzzer(fuzzerOpts, pool());
            return fuzzer.fuzzRow(type);
        }

        std::vector<RowVectorPtr> create_vectors(
            const RowTypePtr &type,
            uint64_t byteSize,
            const VectorFuzzer::Options &fuzzerOpts) {
            VectorFuzzer fuzzer(fuzzerOpts, pool());
            uint64_t totalSize{0};
            std::vector<RowVectorPtr> vectors;
            while (totalSize < byteSize) {
                vectors.push_back(fuzzer.fuzzInputRow(type));
                totalSize += vectors.back()->estimate_flat_size();
            }
            return vectors;
        }

        std::vector<RowVectorPtr>
        create_vectors(const RowTypePtr &type, size_t vectorSize, uint64_t byteSize) {
            return create_vectors(type, byteSize, {.vectorSize = vectorSize});
        }

        std::vector<RowVectorPtr> create_vectors(
            uint32_t numVectors,
            const RowTypePtr &type,
            const VectorFuzzer::Options &fuzzerOpts = {}) {
            VectorFuzzer fuzzer(fuzzerOpts, pool());
            std::vector<RowVectorPtr> vectors;
            vectors.reserve(numVectors);
            for (int i = 0; i < numVectors; ++i) {
                vectors.emplace_back(fuzzer.fuzzRow(type));
            }
            return vectors;
        }

        /// Splits input vector into 'n' vectors evenly. Input vector must have at
        /// least 'n' rows.
        /// @return 'n' vectors
        std::vector<RowVectorPtr> split(const RowVectorPtr &vector, int32_t n = 2);

        /// Returns a one element ArrayVector with 'elements' as elements of array at
        /// 0.
        VectorPtr as_array(VectorPtr elements);

        template<typename T>
        FlatVectorPtr<EvalType<T> > make_flat_vector(
            vector_size_t size,
            std::function<T(vector_size_t /*row*/)> value_at,
            std::function<bool(vector_size_t /*row*/)> is_null_at = nullptr,
            const TypePtr &type = CppToType<T>::create()) {
            return vectorMaker_.flat_vector<T>(size, value_at, is_null_at, type);
        }

        template<typename T>
        FlatVectorPtr<EvalType<T> > make_flat_vector(
            const std::vector<T> &data,
            const TypePtr &type = CppToType<T>::create()) {
            return vectorMaker_.flat_vector<T>(data, type);
        }

        template<typename T>
        FlatVectorPtr<EvalType<T> > make_nullable_flat_vector(
            const std::vector<std::optional<T> > &data,
            const TypePtr &type = CppToType<T>::create()) {
            return vectorMaker_.flat_vector_nullable(data, type);
        }

        template<typename T, int TupleIndex, typename TupleType>
        FlatVectorPtr<T> make_flat_vector(
            const std::vector<TupleType> &data,
            const TypePtr &type = CppToType<T>::create()) {
            return vectorMaker_.flat_vector<T, TupleIndex, TupleType>(data, type);
        }

        template<typename T>
        FlatVectorPtr<T> make_flat_vector(size_t size) {
            return vectorMaker_.flat_vector<T>(size);
        }

        template<typename T>
        FlatVectorPtr<T> make_flat_vector(size_t size, const TypePtr &type) {
            return vectorMaker_.flat_vector<T>(size, type);
        }

        template<typename T>
        FlatVectorPtr<T> make_all_null_flat_vector(vector_size_t size) {
            return vectorMaker_.all_null_flat_vector<T>(size);
        }

        // Convenience function to create arrayVectors (vector of arrays) based on
        // input values from nested std::vectors. The underlying elements are
        // non-nullable.
        //
        // Example:
        //   auto array_vector = make_array_vector<int64_t>({
        //       {1, 2, 3, 4, 5},
        //       {},
        //       {1, 2, 3},
        //   });
        //   EXPECT_EQ(3, array_vector->size());
        template<typename T>
        ArrayVectorPtr make_array_vector(
            const std::vector<std::vector<T> > &data,
            const TypePtr &elementType = CppToType<T>::create()) {
            return vectorMaker_.array_vector<T>(data, elementType);
        }

        ArrayVectorPtr make_all_null_array_vector(
            vector_size_t size,
            const TypePtr &elementType) {
            return vectorMaker_.all_null_array_vector(size, elementType);
        }

        // Create an ArrayVector<ROW> from nested std::vectors of variants.
        // Example:
        //   auto array_vector = make_array_of_row_vector({
        //       {variant::row({1, "red"}), variant::row({1, "blue"})},
        //       {},
        //       {variant::row({3, "green"})},
        //   });
        //   EXPECT_EQ(3, array_vector->size());
        //
        // Use variant(TypeKind::ROW) to specify a null array element.
        ArrayVectorPtr make_array_of_row_vector(
            const RowTypePtr &row_type,
            const std::vector<std::vector<variant> > &data) {
            return vectorMaker_.array_of_row_vector(row_type, data);
        }

        template<typename TupleT>
        ArrayVectorPtr make_array_of_row_vector(
            const std::vector<std::vector<std::optional<TupleT> > > &data,
            const RowTypePtr &row_type) {
            return vectorMaker_.array_of_row_vector(data, row_type);
        }

        // Create an ArrayVector<ArrayVector<T>> from nested std::vectors of values.
        // Example:
        //   using innerArrayType = std::vector<std::optional<int64_t>>;
        //   using outerArrayType =
        //       std::vector<std::optional<std::vector<std::optional<int64_t>>>>;
        //
        //   innerArrayType a{1, 2, 3};
        //   innerArrayType b{4, 5};
        //   innerArrayType c{6, 7, 8};
        //   outerArrayType row1{{a}, {b}};
        //   outerArrayType row2{{a}, {c}};
        //   outerArrayType row3{{{}}};
        //   outerArrayType row4{{{std::nullopt}}};
        //   auto array_vector = make_nullable_nested_array_vector<int64_t>(
        //       {{row1}, {row2}, {row3}, {row4}, std::nullopt});
        //
        //   EXPECT_EQ(5, array_vector->size());
        template<typename T>
        ArrayVectorPtr make_nullable_nested_array_vector(
            const std::vector<std::optional<
                std::vector<std::optional<std::vector<std::optional<T> > > > > > &data) {
            vector_size_t size = data.size();
            BufferPtr offsets = allocateOffsets(size, pool());
            BufferPtr sizes = allocateSizes(size, pool());
            BufferPtr nulls = allocate_nulls(size, pool());

            auto rawOffsets = offsets->asMutable<vector_size_t>();
            auto rawSizes = sizes->asMutable<vector_size_t>();
            auto rawNulls = nulls->asMutable<uint64_t>();

            // Flatten the outermost layer of std::vector of the input, and populate
            // the sizes and offsets for the top-level array vector.
            std::vector<std::optional<std::vector<std::optional<T> > > > flattenedData;
            vector_size_t i = 0;
            for (const auto &vector: data) {
                if (!vector.has_value()) {
                    bits::set_null(rawNulls, i, true);
                    rawSizes[i] = 0;
                    rawOffsets[i] = (i == 0) ? 0 : rawOffsets[i - 1] + rawSizes[i - 1];
                } else {
                    flattenedData.insert(
                        flattenedData.end(), vector->begin(), vector->end());

                    rawSizes[i] = vector->size();
                    rawOffsets[i] = (i == 0) ? 0 : rawOffsets[i - 1] + rawSizes[i - 1];
                }
                ++i;
            }

            // Create the underlying vector.
            auto baseArray = make_nullable_array_vector<T>(flattenedData);

            // Build and return a second-level of ArrayVector on top of baseArray.
            return std::make_shared<ArrayVector>(
                pool(),
                ARRAY(ARRAY(CppToType<T>::create())),
                nulls,
                size,
                offsets,
                sizes,
                baseArray,
                0);
        }

        // Create an ArrayVector<MapVector<TKey, TValue>> from nested std::vectors of
        // pairs. Example:
        //   using S = StringView;
        //   using P = std::pair<int64_t, std::optional<S>>;
        //   std::vector<P> a {P{1, S{"red"}}, P{2, S{"blue"}}, P{3, S{"green"}}};
        //   std::vector<P> b {P{1, S{"yellow"}}, P{2, S{"orange"}}};
        //   std::vector<P> c {P{1, S{"red"}}, P{2, S{"yellow"}}, P{3, S{"purple"}}};
        //   std::vector<std::vector<std::vector<P>>> data = {{a, b, b}, {b, c}, {c,
        //   a, c}};
        //   auto array_vector = make_array_of_map_vector<int64_t, S>(data);
        //
        //   EXPECT_EQ(3, array_vector->size());
        template<typename TKey, typename TValue>
        ArrayVectorPtr make_array_of_map_vector(
            const std::vector<
                std::vector<std::vector<std::pair<TKey, std::optional<TValue> > > > > &
            data) {
            vector_size_t size = data.size();
            BufferPtr offsets = AlignedBuffer::allocate<vector_size_t>(size, pool());
            BufferPtr sizes = AlignedBuffer::allocate<vector_size_t>(size, pool());

            auto rawOffsets = offsets->asMutable<vector_size_t>();
            auto rawSizes = sizes->asMutable<vector_size_t>();

            // Flatten the outermost std::vector of the input and populate the sizes and
            // offsets for the top-level array vector.
            std::vector<std::vector<std::pair<TKey, std::optional<TValue> > > >
                    flattenedData;
            vector_size_t i = 0;
            for (const auto &vector: data) {
                flattenedData.insert(flattenedData.end(), vector.begin(), vector.end());

                rawSizes[i] = vector.size();
                rawOffsets[i] = (i == 0) ? 0 : rawOffsets[i - 1] + rawSizes[i - 1];

                ++i;
            }

            // Create the underlying map vector.
            auto baseVector = make_map_vector<TKey, TValue>(flattenedData);

            // Build and return a second-level of array vector on top of baseVector.
            return std::make_shared<ArrayVector>(
                pool(),
                ARRAY(MAP(CppToType<TKey>::create(), CppToType<TValue>::create())),
                BufferPtr(nullptr),
                size,
                offsets,
                sizes,
                baseVector,
                0);
        }

        template<typename TKey, typename TValue>
        ArrayVectorPtr make_array_of_map_vector(
            const std::vector<std::vector<
                std::optional<std::vector<std::pair<TKey, std::optional<TValue> > > > > > &
            data) {
            vector_size_t size = data.size();
            BufferPtr offsets = AlignedBuffer::allocate<vector_size_t>(size, pool());
            BufferPtr sizes = AlignedBuffer::allocate<vector_size_t>(size, pool());

            auto rawOffsets = offsets->asMutable<vector_size_t>();
            auto rawSizes = sizes->asMutable<vector_size_t>();

            // Flatten the outermost std::vector of the input and populate the sizes and
            // offsets for the top-level array vector.
            std::vector<
                        std::optional<std::vector<std::pair<TKey, std::optional<TValue> > > > >
                    flattenedData;
            vector_size_t i = 0;
            for (const auto &vector: data) {
                flattenedData.insert(flattenedData.end(), vector.begin(), vector.end());
                rawSizes[i] = vector.size();
                rawOffsets[i] = (i == 0) ? 0 : rawOffsets[i - 1] + rawSizes[i - 1];
                ++i;
            }

            // Create the underlying map vector.
            auto baseVector = make_nullable_map_vector<TKey, TValue>(flattenedData);

            // Build and return a second-level of array vector on top of baseVector.
            return std::make_shared<ArrayVector>(
                pool(),
                ARRAY(MAP(CppToType<TKey>::create(), CppToType<TValue>::create())),
                BufferPtr(nullptr),
                size,
                offsets,
                sizes,
                baseVector,
                0);
        }

        // Convenience function to create arrayVectors (vector of arrays) based on
        // input values from nested std::vectors. The underlying array elements are
        // nullable.
        //
        // Example:
        //   auto array_vector = make_nullable_array_vector<int64_t>({
        //       {1, 2, std::nullopt, 4},
        //       {},
        //       {std::nullopt},
        //   });
        //   EXPECT_EQ(3, array_vector->size());
        template<typename T>
        ArrayVectorPtr make_nullable_array_vector(
            const std::vector<std::vector<std::optional<T> > > &data,
            const TypePtr &type = ARRAY(CppToType<T>::create())) {
            std::vector<std::optional<std::vector<std::optional<T> > > > convData;
            convData.reserve(data.size());
            for (auto &array: data) {
                convData.push_back(array);
            }
            return vectorMaker_.array_vector_nullable<T>(convData, type);
        }

        // Just like make_nullable_array_vector above, but allows specifying null arrays.
        //
        //  Example:
        //   auto array_vector = make_nullable_array_vector<int64_t>({
        //       {{1, 2, std::nullopt, 4}},
        //       {{}},
        //       std::nullopt,
        //   });
        template<typename T>
        ArrayVectorPtr make_nullable_array_vector(
            const std::vector<std::optional<std::vector<std::optional<T> > > > &data,
            const TypePtr &arrayType = ARRAY(CppToType<T>::create())) {
            return vectorMaker_.array_vector_nullable<T>(data, arrayType);
        }

        /// Creates an ArrayVector from a list of JSON arrays.
        ///
        /// JSON arrays can represent a null array, an empty array or array with null
        /// elements.
        ///
        /// Examples:
        ///  [1, 2, 3]
        ///  [1, 2, null, 4]
        ///  [null, null]
        ///  [] - empty array
        ///  null - null array
        ///
        /// @tparam T Type of array elements. Must be an integer: int8_t, int16_t,
        /// int32_t, int64_t.
        /// @param jsonArrays A list of JSON arrays. JSON array cannot be an empty
        /// string.
        template<typename T>
        ArrayVectorPtr make_array_vector_from_json(
            const std::vector<std::string> &jsonArrays,
            const TypePtr &arrayType = ARRAY(CppToType<T>::create())) {
            return vectorMaker_.array_vector_from_json<T>(jsonArrays, arrayType);
        }

        template<typename T>
        ArrayVectorPtr make_array_vector(
            vector_size_t size,
            std::function<vector_size_t(vector_size_t /* row */)> sizeAt,
            std::function<T(vector_size_t /* idx */)> value_at,
            std::function<bool(vector_size_t /* row */)> is_null_at = nullptr,
            std::function<bool(vector_size_t /* idx */)> valueIsNullAt = nullptr,
            const TypePtr &arrayType = ARRAY(CppToType<T>::create())) {
            return vectorMaker_.array_vector<T>(
                size, sizeAt, value_at, is_null_at, valueIsNullAt, arrayType);
        }

        /// Similar to make_array_vector_from_json. Creates an ArrayVector from list of
        /// JSON arrays of arrays.
        /// @tparam T Type of array elements. Must be an integer: int8_t, int16_t,
        /// int32_t, int64_t.
        /// @param jsonArrays A list of JSON arrays. JSON array cannot be an empty
        /// string.
        /// @param arrayType type of array elements.
        template<typename T>
        ArrayVectorPtr make_nested_array_vector_from_json(
            const std::vector<std::string> &jsonArrays,
            const TypePtr &arrayType = ARRAY(CppToType<T>::create())) {
            return vectorMaker_.nested_array_vector_from_json<T>(jsonArrays, arrayType);
        }

        template<typename T>
        ArrayVectorPtr make_array_vector(
            vector_size_t size,
            std::function<vector_size_t(vector_size_t /* row */)> sizeAt,
            std::function<T(vector_size_t /* row */, vector_size_t /* idx */)>
            value_at,
            std::function<bool(vector_size_t /*row */)> is_null_at = nullptr,
            const TypePtr &arrayType = ARRAY(CppToType<T>::create())) {
            return vectorMaker_.array_vector<T>(
                size, sizeAt, value_at, is_null_at, arrayType);
        }

        // Convenience function to create vector from a base vector.
        // The size of the arrays is computed from the difference of offsets.
        // An optional vector of nulls can be passed to specify null rows.
        // The offset for a null value must match previous offset
        // i.e size computed should be zero.
        // Example:
        //   auto array_vector = make_array_vector({0, 2 ,2}, elementVector, {1});
        //
        //   creates an array vector with array at index 1 as null.
        // You can make higher order ArrayVectors (i.e array of arrays etc), by
        // repeatedly calling this function and passing in resultant ArrayVector
        // and appropriate offsets.
        ArrayVectorPtr make_array_vector(
            const std::vector<vector_size_t> &offsets,
            const VectorPtr &elementVector,
            const std::vector<vector_size_t> &nulls = {}) {
            return vectorMaker_.array_vector(offsets, elementVector, nulls);
        }

        template<typename TKey, typename TValue>
        MapVectorPtr make_map_vector(
            vector_size_t size,
            std::function<vector_size_t(vector_size_t /* row */)> sizeAt,
            std::function<TKey(vector_size_t /* idx */)> keyAt,
            std::function<TValue(vector_size_t /* idx */)> value_at,
            std::function<bool(vector_size_t /*row */)> is_null_at = nullptr,
            std::function<bool(vector_size_t /*row */)> valueIsNullAt = nullptr,
            const TypePtr &type =
                    MAP(CppToType<TKey>::create(), CppToType<TValue>::create())) {
            return vectorMaker_.map_vector<TKey, TValue>(
                size, sizeAt, keyAt, value_at, is_null_at, valueIsNullAt, type);
        }

        // Create map vector from nested std::vector representation.
        template<typename TKey, typename TValue>
        MapVectorPtr make_map_vector(
            const std::vector<std::vector<std::pair<TKey, std::optional<TValue> > > > &
            maps,
            const TypePtr &mapType =
                    MAP(CppToType<TKey>::create(), CppToType<TValue>::create())) {
            return vectorMaker_.map_vector(maps, mapType);
        }

        // Create nullabe map vector from nested std::vector representation.
        template<typename TKey, typename TValue>
        MapVectorPtr make_nullable_map_vector(
            const std::vector<std::optional<
                std::vector<std::pair<TKey, std::optional<TValue> > > > > &maps,
            const TypePtr &mapType =
                    MAP(CppToType<TKey>::create(), CppToType<TValue>::create())) {
            return vectorMaker_.map_vector(maps, mapType);
        }

        /// Creates a MapVector from a list of JSON maps.
        ///
        /// JSON maps can represent a null map, an empty map or a map with null
        /// values. Null keys are not allowed.
        ///
        /// Examples:
        ///  {1: 10, 2: 20, 3: 30}
        ///  {1: 10, 2: 20, 3: null, 4: 40}
        ///  {1: null, 2: null}
        ///  {} - empty map
        ///  null - null map
        ///
        /// @tparam K Type of map keys. Must be an integer: int8_t, int16_t,
        /// int32_t, int64_t.
        /// @tparam V Type of map value. Can be an integer or a floating point number.
        /// @param jsonMaps A list of JSON maps. JSON map cannot be an empty
        /// string.
        template<typename K, typename V>
        MapVectorPtr make_map_vector_from_json(
            const std::vector<std::string> &jsonMaps,
            const TypePtr &mapType =
                    MAP(CppToType<K>::create(), CppToType<V>::create())) {
            return vectorMaker_.map_vector_from_json<K, V>(jsonMaps, mapType);
        }

        // Convenience function to create vector from vectors of keys and values.
        // The size of the maps is computed from the difference of offsets.
        // The sizes of the key and value vectors must be equal.
        // An optional vector of nulls can be passed to specify null rows.
        // The offset for a null value must match previous offset
        // i.e size computed should be zero.
        // Example:
        //   auto map_vector = make_map_vector({0, 2 ,2}, keyVector, value_vector, {1});
        //
        //   creates a map vector with map at index 1 as null.
        // You can make higher order MapVectors (i.e maps with maps as values etc),
        // by repeatedly calling this function and passing in resultant MapVector
        // and appropriate offsets.
        MapVectorPtr make_map_vector(
            const std::vector<vector_size_t> &offsets,
            const VectorPtr &keyVector,
            const VectorPtr &value_vector,
            const std::vector<vector_size_t> &nulls = {}) {
            return vectorMaker_.map_vector(offsets, keyVector, value_vector, nulls);
        }

        MapVectorPtr make_all_null_map_vector(
            vector_size_t size,
            const TypePtr &keyType,
            const TypePtr &valueType) {
            return vectorMaker_.all_null_map_vector(size, keyType, valueType);
        }

        VectorPtr make_constant(const variant &value, vector_size_t size) {
            return BaseVector::create_constant(value.inferType(), value, size, pool());
        }

        template<typename T>
        VectorPtr make_constant(
            T value,
            vector_size_t size,
            const TypePtr &type = CppToType<EvalType<T> >::create()) {
            return std::make_shared<ConstantVector<EvalType<T> > >(
                pool(), size, false, type, std::move(value));
        }

        template<typename T>
        VectorPtr make_constant(
            const std::optional<T> &value,
            vector_size_t size,
            const TypePtr &type = CppToType<EvalType<T> >::create()) {
            return std::make_shared<ConstantVector<EvalType<T> > >(
                pool(),
                size,
                /*isNull=*/!value.has_value(),
                type,
                value ? EvalType<T>(*value) : EvalType<T>(),
                SimpleVectorStats<EvalType<T> >{},
                sizeof(EvalType<T>));
        }

        /// Create constant vector of type ROW from a Variant.
        VectorPtr make_constant_row(
            const RowTypePtr &row_type,
            variant value,
            vector_size_t size) {
            return vectorMaker_.constant_row(row_type, value, size);
        }

        /// Create constant vector of type ARRAY from a std::vector.
        template<typename T>
        VectorPtr make_constant_array(vector_size_t size, const std::vector<T> &data) {
            return BaseVector::wrap_in_constant(
                size, 0, vectorMaker_.array_vector<T>({data}));
        }

        VectorPtr make_null_constant(TypeKind typeKind, vector_size_t size) {
            return BaseVector::create_null_constant(
                createType(typeKind, {}), size, pool());
        }

        BufferPtr make_indices(
            vector_size_t size,
            std::function<vector_size_t(vector_size_t)> indexAt) const;

        BufferPtr make_indices(const std::vector<vector_size_t> &indices) const;

        BufferPtr make_odd_indices(vector_size_t size);

        BufferPtr make_even_indices(vector_size_t size);

        BufferPtr make_indices_in_reverse(vector_size_t size) {
            return ::kumo::pollux::make_indices_in_reverse(size, pool());
        }

        BufferPtr make_nulls(
            vector_size_t size,
            std::function<bool(vector_size_t /*row*/)> is_null_at);

        /// Creates a null buffer from a vector of booleans.
        BufferPtr make_nulls(const std::vector<bool> &values);

        static VectorPtr
        wrap_in_dictionary(BufferPtr indices, vector_size_t size, VectorPtr vector);

        static VectorPtr wrap_in_dictionary(BufferPtr indices, VectorPtr vector);

        template<typename T = BaseVector>
        static std::shared_ptr<T> flatten(const VectorPtr &vector) {
            return VectorMaker::flatten<T>(vector);
        }

        // Convenience function to create a vector of type Map(K, ARRAY(K)).
        // Supports null keys, and values and even null elements.
        // Example:
        //    create_map_of_arrays_vector<int64_t>(
        //      {{{1, std::nullopt}},
        //       {{2, {{4, 5, std::nullopt}}}},
        //       {{std::nullopt, {{7, 8, 9}}}}});
        template<typename K, typename V>
        VectorPtr create_map_of_arrays_vector(
            std::vector<std::map<
                std::optional<K>,
                std::optional<std::vector<std::optional<V> > > > > maps) {
            std::vector<std::optional<K> > keys;
            std::vector<std::optional<std::vector<std::optional<V> > > > values;
            for (auto &map: maps) {
                for (const auto &[key, value]: map) {
                    keys.push_back(key);
                    values.push_back(value);
                }
            }

            auto mapValues = make_nullable_array_vector(values);
            auto mapKeys = make_nullable_flat_vector<K>(keys);
            auto size = maps.size();

            auto offsets = AlignedBuffer::allocate<vector_size_t>(size, pool_.get());
            auto sizes = AlignedBuffer::allocate<vector_size_t>(size, pool_.get());

            auto rawOffsets = offsets->template asMutable<vector_size_t>();
            auto rawSizes = sizes->template asMutable<vector_size_t>();

            vector_size_t offset = 0;
            for (vector_size_t i = 0; i < size; i++) {
                rawSizes[i] = maps[i].size();
                rawOffsets[i] = offset;
                offset += maps[i].size();
            }

            return std::make_shared<MapVector>(
                pool_.get(),
                MAP(CppToType<K>::create(), ARRAY(CppToType<V>::create())),
                nullptr,
                size,
                offsets,
                sizes,
                mapKeys,
                mapValues);
        }

        memory::MemoryPool *pool() const {
            return pool_.get();
        }

        // Create LazyVector that produces a flat vector and asserts that is is being
        // loaded for a specific set of rows.
        template<typename T>
        std::shared_ptr<LazyVector> make_lazy_flat_vector(
            vector_size_t size,
            std::function<T(vector_size_t /*row*/)> value_at,
            std::function<bool(vector_size_t /*row*/)> is_null_at =
                    [](vector_size_t row) { return false; },
            std::optional<vector_size_t> expectedSize = std::nullopt,
            const std::optional<
                std::function<vector_size_t(vector_size_t /*index*/)> > &
                    expectedRowAt = std::nullopt) {
            return std::make_shared<LazyVector>(
                pool(),
                CppToType<T>::create(),
                size,
                std::make_unique<SimpleVectorLoader>([=](RowSet rows) {
                    if (expectedSize.has_value()) {
                        POLLUX_CHECK_EQ(rows.size(), *expectedSize);
                    }
                    if (expectedRowAt.has_value()) {
                        for (auto i = 0; i < rows.size(); i++) {
                            POLLUX_CHECK_EQ(rows[i], (*expectedRowAt)(i));
                        }
                    }
                    return make_flat_vector<T>(size, value_at, is_null_at);
                }));
        }

        VectorPtr wrap_in_lazy_dictionary(VectorPtr vector) {
            return std::make_shared<LazyVector>(
                pool(),
                vector->type(),
                vector->size(),
                std::make_unique<SimpleVectorLoader>([=](RowSet /*rows*/) {
                    auto indices =
                            make_indices(vector->size(), [](auto row) { return row; });
                    return wrap_in_dictionary(indices, vector->size(), vector);
                }));
        }

        std::shared_ptr<memory::MemoryPool> rootPool_{
            memory::memoryManager()->addRootPool()
        };
        std::shared_ptr<memory::MemoryPool> pool_{rootPool_->addLeafChild("leaf")};
        VectorMaker vectorMaker_{pool_.get()};
        std::shared_ptr<melon::Executor> executor_{
            std::make_shared<melon::CPUThreadPoolExecutor>(
                std::thread::hardware_concurrency())
        };
        std::shared_ptr<melon::Executor> spillExecutor_{
            std::make_shared<melon::CPUThreadPoolExecutor>(
                std::thread::hardware_concurrency())
        };
    };
} // namespace kumo::pollux
